package cn.dmp.report

import org.apache.spark.sql.SQLContext
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable.ListBuffer

object RptAreaAnalysisCoreTest {


    def main(args: Array[String]): Unit = {
        // 检验参数 -> 配置文件
        if (args.length != 2) {
            println(
                """
                  |cn.dmp.report.RptAreaAnalysisCore
                  |参数：dataInputPath, outputPath
                """.stripMargin)
            sys.exit()
        }

        val Array(dataInputPath, outputPath) = args

        // sparkcontext
        val sparkConf = new SparkConf().setMaster("local[*]").setAppName("地域报表")
        val sc = new SparkContext(sparkConf)

        // 读取数据 - parquet
        val sqlc = new SQLContext(sc)
        // 数据文件 - dataframe
        val parquet = sqlc.read.parquet(dataInputPath)

        // 算子的方式实现 -> 聚合 reduceByKey(对偶元组)  groupByKey
        parquet.map(row => {

            val list = new ListBuffer[Double]()

            val pName = row.getAs[String]("provincename")
            val cName = row.getAs[String]("cityname")
            val rMode = row.getAs[Int]("requestmode")
            val pNode = row.getAs[Int]("processnode")
            val effect = row.getAs[Int]("iseffective")
            val bill = row.getAs[Int]("isbilling")
            val isbid = row.getAs[Int]("isbid")
            val iswin = row.getAs[Int]("iswin")
            val adorderid = row.getAs[Int]("adorderid")
            val winprice = row.getAs[Double]("winprice")
            val adpayment = row.getAs[Double]("adpayment")

            // 广告请求
            if (rMode == 1) {
                if (pNode == 1) {
                    list.append(1, 0, 0)
                } else if (pNode == 2) {
                    list.append(1, 1, 0)
                } else if (pNode == 3) {
                    list.append(1, 1, 1)
                } else list.append(0, 0, 0)
            } else {
                list.append(0, 0, 0)
            }

            // 广告参与竞价
            if (effect == 1 && bill == 1) {
                if (isbid == 1 && adorderid != 0) list.append(1, 0) else list.append(0,0)
                if (iswin == 1) list.append(0, 1) else list.append(0, 0)
            } else list.append(0, 0) // 参与竞价，竞价成功

            // 展示点击
            if (effect == 1) {
                if (rMode == 2) {
                    list.append(1, 0)
                } else if (rMode == 3) {
                    list.append(0, 1)
                } else list.append(0, 0)
            } else list.append(0, 0)

            // 消费成本
            if (effect == 1 && bill == 1 && iswin == 1) {
                list.append(winprice / 1000, adpayment / 1000)
            } else list.append(0, 0)

            ((pName, cName), list)

        }).reduceByKey((list1, list2) => {
            list1.zip(list2).map(t => t._1 + t._2)
        }).map(t => t._1._1 + "," +t._1._2 + "," + t._2.mkString(","))
          .foreach(println)

        sc.stop()
    }

}
